#! /bin/bash
export LANG=en_US.UTF-8

#WARNLIMIT=-1 		#警告值 超过该值则认为是需要告警
#ERRORLIMIT=-1 		#错误值 超过该值认为是错误

FLAGE=
FLAGEID=
#是否输出FLAGPID
FLAGEOPID=1
#是否输出RUNTIME
RUNTIME=1
#是否输出STARTTIME
STARTTIME=0
LAG=0
LAGMIN=
LAGMAX=
TOPICFILE=			#topic列表临时保存文件
TOPICFILEAPPEN="kafka_temp"				#topic列表临时保存文件后缀
GROUPSTR=			#受检的group字符串信息
GROUPLIST=			#受检的group列表
TOPLIST=			#受检的toppic列表
BINDIR=				#kafka工具地址
ZOOKEEPERADDR=		#zookeeper地址

WranResultStr=		#最后形成的告警字符串
ErrorResultStr=		#最后形成的错误字符串
DataResultStr=		#性能参数字符串
MSGStr=				#短信告警信息
HEADER="template=freedom Lag告警列表=消费者；TOPPIC；PID；当前值；当前状态；告警阀值；紧急阀值&"
CURNUM=0			#LAG剩余临时变量

function gotErr(){
	if [ "$1" -eq 0 ];then
		echo "mailstatedes=各项指标恢复正常 statedes=各项指标恢复正常|$DataResultStr"
	elif [ "$1" -eq 1 ];then 
		echo "$HEADER$WranResultStr statedes=$MSGStr|$DataResultStr"
	elif [ "$1" -eq 2 ];then 
		if [[ "${WranResultStr}" != "" ]]; then
			echo "$HEADER$WranResultStr&$ErrorResultStr statedes=$MSGStr|$DataResultStr"
			else
			echo "$HEADER$ErrorResultStr statedes=$MSGStr|$DataResultStr"
		fi
		
	elif [ "$1" -eq 3 ];then 
		echo "mailstatedes=无法获取到指标,疑似组件故障,请确认并请检查监控脚本和运维平台配置 statedes=无法获取到指标,疑似组件故障,请确认并请检查监控脚本和运维平台配置|$DataResultStr"
	fi

	exit $1
}

function isValidResult(){
	isValidStr=$(cat $TOPICFILE | grep -i "except" | wc -l)

	#zookeeper链接失败
	if [[ "$isValidStr" -gt 0 ]];then
		if [ ! -d "./except" ];then
			mkdir except
		fi
		cp $TOPICFILE "./except/"$TOPICFILE"_except_"$(date +%s)
		gotErr 3
	fi
}

#ps -eo pid,etime 是查询进程运行时间
function getRuntime(){
	CURNUM=$(ps -eo pid,etime | grep $FLAGEID |awk '{if($1=='$FLAGEID') print $2}')
}

#ps -eo pid,lstart是查询进程的开始时间
function getStarttime(){
	CURNUM=$(ps -eo pid,lstart | grep $FLAGEID |awk '{if($1=='$FLAGEID') print $2 ,$3 ,$4 ,$5 ,$6}')
}

function getFlagNum(){
	groupname=$1
	$BINDIR/kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --group $groupname --zkconnect $ZOOKEEPERADDR 1>"$TOPICFILE" 2>&1
	#查询结果是否合法
	isValidResult
	OLDIFS=$IFS
	IFS=$'\n'
	for line in `cat $TOPICFILE|grep -v Group`
	do
		pid=$(echo $line|awk '{print $3}')
		CURNUM=$(echo $line|awk '{print $6}')
		toppicname=$(echo $line|awk '{print $2}')
		parseFileds $groupname $toppicname $pid $LAGMIN $LAGMAX
	done
	IFS=$OLDIFS
	
	if [ "$FLAGEOPID" -eq 1 ];then
			CURNUM=$FLAGEID
			parseFileds "FLAGEPID" $CURNUM $CURNUM
	fi
	
	if [ "$RUNTIME" -eq 1 ];then
			getRuntime
			parseTimeFileds "RUNTIME" $RUNTIME $RUNTIME		
	fi
	if [ "$STARTTIME" -eq 1 ];then
			getStarttime
			parseTimeFileds "STARTTIME" $STARTTIME $STARTTIME		
	fi
	
	
}
function parseTimeFileds(){
	val1=$(echo "$CURNUM $2")
	val2=$(echo "$CURNUM $3")
  DataResultStr="$DataResultStr $1=$CURNUM;$2;$3;0;0"
}
#依据传入的参数值来判断我们要监控的服务是什么
#然后通过各个函数获取到具体的值
#值之间的
function analysisStat(){
	if [ "$LAG" -eq 1 ];then
		getGroupList
		for group in $GROUPLIST
		do
			getFlagNum $group
		done
	fi
}

#调用kafka工具脚本获取toppic列表
function getToppicList(){
	$BINDIR/kafka-topics.sh --list --zookeeper $ZOOKEEPERADDR 1>"$TOPICFILE" 2>&1
	
	#查询结果是否合法
	isValidResult
	index=0
	for line in `cat $TOPICFILE`
	do
		TOPICLIST[$index]=$line
		((index++))
	done
}

#通过参数传入获取受检group列表
function getGroupList(){
	OLD_IFS="$IFS" 
	IFS="," 
	GROUPLIST=($GROUPSTR)
	IFS="$OLD_IFS"
}

#CURNUM			当前值
#WARNLIMIT 		告警的上限
#ERRORLIMIT 	异常的上限
#判断以上三个值并拼接结果字符串
function parseFileds(){
	val1=$(echo "$CURNUM $4"| awk '{if($1<=$2){print 1}else{print 0}}')
	val2=$(echo "$CURNUM $5"| awk '{if($1<=$2){print 1}else{print 0}}')

	DataResultStr="$DataResultStr $1_$2_$3_Lag=$CURNUM;$4;$5;0;0"
	StateStr=
	if [ $val1 -eq 1 ];then
		return 0
	elif [[ $val1 -eq 0 && $val2 -eq 1 ]];then
		if [[ ${WranResultStr} != "" ]]; then
			WranResultStr="${WranResultStr}&"
		fi

		WranResultStr="${WranResultStr}$1；$2；$3；$CURNUM；告警；$4；$5"
		
		if [[ ${MSGStr} != "" ]]; then
			MSGStr="${MSGStr}；"
		fi
		MSGStr="${MSGStr}消费者-$1的Toppic-$2日志在PID-$3日志消费剩余量为$CURNUM超过告警阀值$4"
		return 1
	else
		if [[ ${ErrorResultStr} != "" ]]; then
			ErrorResultStr="${ErrorResultStr}&"
		fi
		ErrorResultStr="${ErrorResultStr}$1；$2；$3；$CURNUM；紧急；$4；$5"
		
		if [[ ${MSGStr} != "" ]]; then
			MSGStr="${MSGStr}；"
		fi
		MSGStr="${MSGStr}消费者-$1的Toppic-$2日志在PID-$3日志消费剩余量为$CURNUM超过紧急阀值$5"
		return 2
	fi
}

#分析结果 给出给出状态
function analysisResult(){
	if [[ "$WranResultStr" == "" && "$ErrorResultStr" == "" ]];then
		gotErr 0
	elif [[ "$ErrorResultStr" == "" ]];then
		gotErr 1
	else
		gotErr 2
	fi
}

#输出错误信息并且退出程序
function showHelp(){
	if [ "$1" != "" ];then
		echo "请输入$1"
	fi
	echo "check_kafka.sh 可以监听本地的kafka的状态 参数如下"
	echo "check_kafka.sh [-w LAG剩余量] [-c LAG剩余量] -g <GroupId> -e <kafka bin path> -z <zookeeper address>"
	echo "-w 表示监控警告的值"
	echo "-c 表示异常值"
	echo "-g 监控的group列表"
	echo "-e kafka bin目录"
	echo "-z zookeeper地址"
	exit 1
}


#检查参数是不是数字
function isNumber(){
	isNum=$(echo "$1" | awk '{print($0~/^([0-9])+[\.]?([0-9])?$/)?1:0}')
	return $isNum
}

function isLarge(){

	isNumber $1
	if [[ $? -eq 0 ]]; then
		return 1
	fi

	isNumber $2
	if [[ $? -eq 0 ]]; then
		return 1
	fi

	isLarge=$(echo "$1 $2"| awk '{if($1>=$2){print 1}else{print 0}}')
	if [ $isLarge -eq 1 ];then
		return 0
	else
		return 1
	fi
}

function checkARG(){
	if [[ "$2" != "" && "$3" != "" ]];then
		isLarge "$3" "$2"
		if [[ $? -eq 1 ]]; then
			return 1
		else
			$1=1
		fi
	fi
	return 0
}

#检查参数是否合法
#即参数是否是小数和整数
#对于守护进程的选择 这个值可以不验证
function checkALL(){
	if [[ "$LAGMIN" != "" || "$LAGMAX" != "" ]];then

		if [[ "$LAGMIN" == "" || "$LAGMAX" == "" ]]; then
			return 0
		fi
		isLarge "$LAGMAX" "$LAGMIN"
		if [[ $? -eq 1 ]]; then
			return 1
		else
			LAG=1
		fi
	fi
	return 0
}

function parseARG(){
	#it mean warning
	if [ $1 -eq 1 ];then
 		LAGMIN=$(echo $2 | awk -F ',' '{print $1}' | sed "s/[^0-9]//g")
 	else
 		LAGMAX=$(echo $2 | awk -F ',' '{print $1}' | sed "s/[^0-9]//g")
 	fi
}

while getopts "w:c:g:e:z:h:" arg
do
	case $arg in
		w)
			parseARG 1 "$OPTARG"
			;;
		c)
			parseARG 2 "$OPTARG"
			;;
		h)
			showHelp
			;;
		g)
			GROUPSTR=$OPTARG
			;;
		e)
			BINDIR=$OPTARG
			;;
		z)
			ZOOKEEPERADDR=$OPTARG
			
			;;
		?) 
			showHelp
			;;
	esac
done

checkALL
TOPICFILE=$(echo "$ZOOKEEPERADDR" | sed -s 's/\//_/g')$GROUPSTR"_"$TOPICFILEAPPEN
if [[ $? == 1 ]];then
	showHelp "正确参数"
fi


#插件正题部分 
#启动
#分析

analysisStat
analysisResult
